Atividade Integradora¶
- Big Data e Computação em Nuvem / Modelos Preditivos Avançados
Insper - Jul/2020 - São Paulo, Brasil
- Autores:
- Marcelo Franceschini
- Rafael Costa
- Viviane Sanchez
Objetivo:¶
Recomendar um estabelecimento bem avaliado que o usuário ainda não foi
Ambiente Spark¶
In [1]:
import findspark as fs
from pyspark.sql import SparkSession
In [98]:
import matplotlib.pyplot as plt
from wordcloud import WordCloud, ImageColorGenerator
import os
from pyspark.sql import SparkSession
# from pyspark.sql.types import *
from pyspark.sql import functions as f
from pyspark.sql.window import Window
from pyspark.ml.feature import StopWordsRemover
from pyspark.sql.functions import col, count, explode, split, array, concat_ws
from pyspark.sql.types import ArrayType, IntegerType
import pandas as pd
import seaborn as sns
from tqdm.notebook import tqdm
sns.set(style="ticks", palette="pastel")
%matplotlib inline
In [3]:
#MAC Local (Viviane)
spark_location='/Users/vivi/server/spark' # Set your own
java8_location= '/Library/Java/JavaVirtualMachines/jdk1.8.0_251.jdk/Contents/Home/' # Set your own
os.environ['JAVA_HOME'] = java8_location
fs.init(spark_home=spark_location)
In [4]:
#datapath = 'C:\\Users\\RuWindows\\Desktop\\PI\\yelp_dataset\\' #Marcelo
#datapath = '../data/yelp' #servidor AWS
datapath = 'data' #Viviane
In [5]:
#files = sorted(os.listdir(datapath))
#files
In [6]:
#!head data/yelp_academic_dataset_review.json
In [7]:
# Spark Session
#spark = SparkSession.builder \
# .master('local[*]') \
# .appName('Integradora Yelp') \
# .config("spark.ui.port", "4060") \
# .getOrCreate()
In [8]:
spark = SparkSession.builder \
.master('local[8]') \
.appName('Yelp Integradora') \
.getOrCreate()
In [9]:
sc = spark.sparkContext
In [10]:
spark#.stop()
Out[10]:
Importando as Bases Origem - Raw¶
In [11]:
usr_raw = spark.read.json(datapath+'/yelp_academic_dataset_user.json')
rv_raw = spark.read.json(datapath+'/yelp_academic_dataset_review.json')
bz_raw = spark.read.json(datapath+'/yelp_academic_dataset_business.json')
tp_raw = spark.read.json(datapath+'/yelp_academic_dataset_tip.json')
ch_raw = spark.read.json(datapath+'/yelp_academic_dataset_checkin.json')
In [12]:
# Visualizando Estrutura
#bz_raw.printSchema()
Tratamento das bases¶
Check-ins¶
- Explosão das datas na base
In [13]:
ch = ch_raw.withColumn('dt', f.explode(f.split(f.col('date'),', ')))
In [14]:
ch_raw.show()
In [15]:
ch.createOrReplaceTempView('ch')
In [16]:
ch1 = spark.sql('''
SELECT business_id, dt,
DATE(dt) AS date,
DAYOFWEEK(dt) AS dow,
MONTH(dt) AS month,
YEAR(dt) AS year,
HOUR(dt) AS hour,
DAY(dt) AS day
FROM ch
''')
In [17]:
ch1.show()
In [18]:
ch1.createOrReplaceTempView('ch1')
In [19]:
ch2 = spark.sql('''
SELECT business_id, year, month, dow, day,
COUNT(date) AS count
FROM ch1
GROUP BY business_id, year, month, dow, day
ORDER BY count DESC
''')
In [20]:
ch2.createOrReplaceTempView('ch2')
Analisando o movimento de um estabelecimento¶
In [21]:
df = spark.sql('''
SELECT *
FROM ch2
WHERE business_id = 'BxCzy1WOVxOrr_G7V4BIvg'
''').toPandas()
In [22]:
df2 = df.groupby(['month','dow'])['count'].sum().unstack('dow').fillna(0)
In [23]:
plt.figure(figsize = (25,7))
sns.heatmap(data = df2, cmap="YlGnBu", annot = True)
Out[23]:
Pelo gráfico, verfica-se que o local é ais movimentado em janeiro, principalmente aos finais de semana. Para a modelagem, o ideal seria utilizar a moda (valor mais frequente) dos check-ins. A média pode não ser representativa, pois o pico de movimento está concentrado aos finais de semana.
Tips¶
In [24]:
tp_raw.createOrReplaceTempView('tp')
In [25]:
tp_raw.show()
- Quantidade de tips por estabelecimento e usuário
In [26]:
tp_usr = spark.sql('''
SELECT user_id,
count(text) AS tips_counter,
sum(compliment_count) as total_compliments
FROM tp
GROUP BY user_id
ORDER BY total_compliments DESC
''')
In [27]:
tp_usr.show()
In [28]:
tp_usr.createOrReplaceTempView('tp_usr')
In [29]:
tp_bz = spark.sql('''
SELECT business_id,
count(text) AS tips_counter,
sum(compliment_count) as total_compliments
FROM tp
GROUP BY business_id
ORDER BY total_compliments DESC
''')
In [30]:
tp_bz.createOrReplaceTempView('tp_bz')
In [31]:
tp_bz.show()
A quantidade total e o alcance das dicas por estabelecimento serão utilizadas na modelagem.
Business¶
Abertura dos atributos e horários para colunas¶
In [32]:
dfs = []
for x in ["hours", "attributes"]:
cols = bz_raw.select(f"{x}.*").columns
for col in cols:
try:
dfs.append(dfs[-1].withColumn(col, f.col(f"{x}.{col}")))
except IndexError:
dfs.append(bz_raw.withColumn(col, f.col(f"{x}.{col}")))
bz = dfs[-1].drop("hours", "attributes")
bz.createOrReplaceTempView("bz")
bz.printSchema()
Limpeza e padronização dos atributos¶
In [33]:
bz1 = spark.sql('''
SELECT *
FROM bz
WHERE city == 'Toronto'
AND is_open == 1
''')
In [34]:
bz1.groupBy('is_open').count().show()
In [35]:
cols = bz_raw.select('attributes.*').columns
Valores únicos em cada coluna de atributos. As colunas que contém listas serão desconsideradas da base para facilitar a análise. As demais serão codificadas conforme a seguir:
- Null/None: 0
- False: 1
- True: 2
As colunas que possuem uma descrição das caracterÃsticas serão tratadas individualmente.
In [36]:
[bz1.groupBy(cols[x]).count().show() for x in range(len(cols))]
Out[36]:
In [37]:
unwanted = ['Ambience','BestNights','BusinessAcceptsBitcoin','BusinessParking','BYOBCorkage','DietaryRestrictions','GoodForMeal','HairSpecializesIn','Music','Open24Hours','RestaurantsCounterService']
In [38]:
bz2 = bz1.drop('Ambience','BestNights','BusinessAcceptsBitcoin','BusinessParking','BYOBCorkage','DietaryRestrictions','GoodForMeal','HairSpecializesIn','Music','Open24Hours','RestaurantsCounterService',
'Sunday','Monday','Tuesday','Wednesday','Thursday','Friday','Saturday','address','postal_code','state','city','is_open')
In [39]:
for col in unwanted:
cols.remove(col)
In [40]:
bz3 = bz2.fillna('0')
In [41]:
bz4 = bz3.replace({'True':'2', 'False':'1','None':'0','null':'0'}, cols)
In [42]:
bz5 = bz4.withColumn('AgesAllowed',f.regexp_replace(f.col('AgesAllowed'), "u'allages'",'0'))
bz6 = bz5.withColumn('AgesAllowed',f.regexp_replace(f.col('AgesAllowed'), "u'19plus'",'19'))
In [43]:
bz7 = bz6.withColumn('Alcohol',f.regexp_replace(f.col('Alcohol'), "'none'|u'none'",'0'))
bz8 = bz7.withColumn('Alcohol',f.regexp_replace(f.col('Alcohol'), "'beer_and_wine'|u'beer_and_wine'",'1'))
bz9 = bz8.withColumn('Alcohol',f.regexp_replace(f.col('Alcohol'), "'full_bar'|u'full_bar'",'2'))
In [44]:
bz10 = bz9.withColumn('NoiseLevel',f.regexp_replace(f.col('NoiseLevel'), "'quiet'|u'quiet'",'1'))
bz11 = bz10.withColumn('NoiseLevel',f.regexp_replace(f.col('NoiseLevel'), "'average'|u'average'",'2'))
bz12 = bz11.withColumn('NoiseLevel',f.regexp_replace(f.col('NoiseLevel'), "'very_loud'|u'very_loud'",'4'))
bz13 = bz12.withColumn('NoiseLevel',f.regexp_replace(f.col('NoiseLevel'), "'loud'|u'loud'",'3'))
In [45]:
bz14 = bz13.withColumn('Smoking',f.regexp_replace(f.col('Smoking'), "'no'|u'no'",'1'))
bz15 = bz14.withColumn('Smoking',f.regexp_replace(f.col('Smoking'), "'outdoor'|u'outdoor'",'2'))
bz16 = bz15.withColumn('Smoking',f.regexp_replace(f.col('Smoking'), "'yes'|u'yes'",'3'))
In [46]:
bz17 = bz16.withColumn('WiFi',f.regexp_replace(f.col('WiFi'), "'no'|u'no'",'1'))
bz18 = bz17.withColumn('WiFi',f.regexp_replace(f.col('WiFi'), "'paid'|u'paid'",'2'))
bz19 = bz18.withColumn('WiFi',f.regexp_replace(f.col('WiFi'), "'free'|u'free'",'3'))
In [47]:
bz20 = bz19.withColumn('RestaurantsAttire',f.regexp_replace(f.col('WiFi'), "'casual'|u'casual'",'1'))
bz21 = bz20.withColumn('RestaurantsAttire',f.regexp_replace(f.col('WiFi'), "'dressy'|u'dressy'",'2'))
bz22 = bz21.withColumn('RestaurantsAttire',f.regexp_replace(f.col('WiFi'), "'formal'|u'formal'",'3'))
In [48]:
#[bz22.groupBy(cols[x]).count().show() for x in range(len(cols))]
In [49]:
bz22.createOrReplaceTempView('bz22')
Join da quantidade de tips por estabelecimento¶
In [50]:
bz23 = spark.sql('''
SELECT A.*,
IFNULL(B.tips_counter,0) AS tips_counter_bz,
IFNULL(B.total_compliments,0) AS total_compliments_bz
FROM bz22 as A
LEFT JOIN tp_bz as B
ON A.business_id = B.business_id
''')
In [51]:
bz23.columns
Out[51]:
In [52]:
bz23.write \
.format('csv') \
.mode('overwrite') \
.option('sep', ',') \
.option('header', True) \
.save('output/yelp_bz.csv')
In [53]:
bz23.createOrReplaceTempView('bz')
Users¶
Seleção e tratamento da variáveis¶
In [54]:
rv_raw.createOrReplaceTempView('rv')
In [55]:
# Transforma variáveis para serem utilizadas na Matriz de Distâncias - H Clust
# Arrays em Numéricas - "id" de friends para friends_count
# Arrays em Numéricas - "ano" de elite para elite_count
# Data em Numéricas do ano - "yelping_since" para "year_since"
from pyspark.sql.functions import col, count, explode, split, array
from pyspark.sql.types import ArrayType, IntegerType
usr_raw1 = usr_raw.withColumn('friends_count', f.size(f.split(f.col('friends'), ' ')))
usr_raw1 = usr_raw1.withColumn('elite_count', f.size(f.split(f.col('elite'), ',')))
usr_raw1 = usr_raw1.withColumn('year_since', f.year('yelping_since'))
#usr_raw1.show()
In [56]:
usr_raw1.createOrReplaceTempView('usr_raw1')
- Seleção variáveis que serão utilizadas na clusterização
In [57]:
usr_base = spark.sql('''
SELECT
A.user_id,
A.average_stars,
A.compliment_cool,
A.compliment_cute,
A.compliment_funny,
A.compliment_hot,
A.compliment_list,
A.compliment_more,
A.compliment_note,
A.compliment_photos,
A.compliment_plain,
A.compliment_profile,
A.compliment_writer,
A.cool,
A.elite_count,
A.fans,
A.friends_count,
A.funny,
A.review_count as review_count_usr,
A.useful,
A.year_since,
B.business_id
FROM usr_raw1 as A LEFT JOIN rv as B
ON A.user_id = B.user_id
''')
In [58]:
usr_base.createOrReplaceTempView('usr_base')
In [59]:
spark.sql('''
SELECT
COUNT(DISTINCT user_id)
FROM usr_base
''').show()
- Seleção de usuários com reviews de business IDs em Toronto, já selecionados anteriormente na base de estabelecimentos
In [60]:
usr_base2 = spark.sql('''
SELECT
A.business_id,
B.*
FROM bz as A
LEFT JOIN usr_base as B
ON A.business_id = B.business_id
''')
In [61]:
usr_base2.createOrReplaceTempView('usr_base2')
In [62]:
spark.sql('''
SELECT
COUNT(DISTINCT user_id)
FROM usr_base2
''').show()
In [63]:
# Identificar User_id Apenas de Usuários que fizeram Reviews em Estabelecimentos Abertos Em Toronto
usr_base3 = spark.sql('''
SELECT
DISTINCT user_id
FROM usr_base2
''')
In [64]:
usr_base3.createOrReplaceTempView('usr_base3')
In [65]:
# Seleciona Apenas de Usuários que fizeram Reviews em Estabelecimentos Abertos Em Toronto
usr_base4 = spark.sql('''
SELECT
A.user_id,
A.average_stars,
A.compliment_cool,
A.compliment_cute,
A.compliment_funny,
A.compliment_hot,
A.compliment_list,
A.compliment_more,
A.compliment_note,
A.compliment_photos,
A.compliment_plain,
A.compliment_profile,
A.compliment_writer,
A.cool,
A.elite_count,
A.fans,
A.friends_count,
A.funny,
A.review_count AS review_count_usr,
A.useful,
A.year_since
FROM usr_raw1 as A LEFT JOIN usr_base3 as B
ON A.user_id = B.user_id
WHERE B.user_id <> ''
''')
In [66]:
usr_base4.createOrReplaceTempView('usr_base4')
- Quantidade inicial de usuários na base:
In [67]:
spark.sql('''
SELECT
COUNT(DISTINCT user_id)
FROM usr_raw1
''').show()
- Quantidade final de usuários na base:
In [68]:
spark.sql('''
SELECT
COUNT(DISTINCT user_id)
FROM usr_base4
''').show()
Join de tips por usuário¶
In [69]:
usr = spark.sql('''
SELECT A.*,
IFNULL(B.tips_counter,0) AS tips_counter,
IFNULL(B.total_compliments,0) AS total_compliments
FROM usr_base4 as A
LEFT JOIN tp_usr as B
ON A.user_id = B.user_id
''')
In [70]:
usr.count()
Out[70]:
In [71]:
usr.write \
.format('csv') \
.mode('overwrite') \
.option('sep', ',') \
.option('header', True) \
.save('output/yelp_usr.csv')
In [72]:
usr.createOrReplaceTempView('usr')
In [73]:
usr.columns
Out[73]:
Análise das bases Business e Users no R¶
Base para treino do modelo¶
In [74]:
usr_clusters = spark.read.csv('output/usr_cluster.csv', header = True)
usr_clusters.createOrReplaceTempView('usr_clusters')
In [75]:
#Juntando número do cluster na base de usuários
usr_clusters2 = spark.sql("""
SELECT A.*,
B.cluster_usr
FROM usr as A
LEFT JOIN usr_clusters as B
ON A.user_id = B.user_id
""")
In [76]:
usr_clusters2.createOrReplaceTempView('usr_clusters2')
In [77]:
#seleção das reviews daqueles usuários
bf1 = spark.sql("""
SELECT
B.*,
A.business_id,
A.stars as stars_rv,
YEAR(A.date) AS year_rv
FROM usr_clusters2 as B
LEFT JOIN rv as A
ON B.user_id = A.user_id
""")
In [78]:
bf1.columns
Out[78]:
In [79]:
bf1.createOrReplaceTempView('bf1')
- Para evitar informações muito desatualizadas na base, serão consideradas apenas reviews a partir de 2017, que corresponde à maior parte da base. de reviews.
In [80]:
rv_df = spark.sql("""
SELECT stars_rv, year_rv, count(stars_rv) as qtde_rv
FROM bf1 as A
GROUP BY stars_rv, year_rv
""").toPandas()
In [81]:
rv_df.groupby(['year_rv','stars_rv'])['qtde_rv'].agg('sum').unstack('stars_rv').plot(kind = 'bar', stacked = True, figsize = (25,6))
Out[81]:
In [82]:
bf2 = spark.sql('''
SELECT A.*,
B.categories,
B.latitude,
B.longitude,
B.name,
B.review_count,
B.stars,
B.AcceptsInsurance,
B.AgesAllowed,
B.Alcohol,
B.BYOB,
B.BikeParking,
B.BusinessAcceptsCreditCards,
B.ByAppointmentOnly,
B.Caters,
B.CoatCheck,
B.Corkage,
B.DogsAllowed,
B.DriveThru,
B.GoodForDancing,
B.GoodForKids,
B.HappyHour,
B.HasTV,
B.NoiseLevel,
B.OutdoorSeating,
B.RestaurantsAttire,
B.RestaurantsDelivery,
B.RestaurantsGoodForGroups,
B.RestaurantsPriceRange2,
B.RestaurantsReservations,
B.RestaurantsTableService,
B.RestaurantsTakeOut,
B.Smoking,
B.WheelchairAccessible,
B.WiFi,
B.tips_counter_bz,
B.total_compliments_bz
FROM bf1 as A
RIGHT JOIN bz as B
ON A.business_id = B.business_id
WHERE year_rv >= 2017
''')
In [83]:
bf2.createOrReplaceTempView('bf2')
In [84]:
spark.sql('''
SELECT
COUNT(DISTINCT business_id)
FROM bf2
''').show()
In [85]:
bf2.columns
Out[85]:
In [86]:
bf2.write \
.format('csv') \
.mode('overwrite') \
.option('sep', ',') \
.option('header', True) \
.save('output/yelp.csv')
Reviews¶
Tratamento do texto¶
In [87]:
#Função para remoção de pontuação e tratamento das palavras
def word_clean(sdf,col,new_col):
rv1 = sdf.withColumn(new_col,f.regexp_replace(f.col(col), "'d", " would"))
rv2 = rv1.withColumn(new_col,f.regexp_replace(f.col(new_col), "'ve", " have"))
rv3 = rv2.withColumn(new_col,f.regexp_replace(f.col(new_col), "'s", " is"))
rv4 = rv3.withColumn(new_col,f.regexp_replace(f.col(new_col), "'re", " are"))
rv5 = rv4.withColumn(new_col,f.regexp_replace(f.col(new_col), "n't", " not"))
rv6 = rv5.withColumn(new_col,f.regexp_replace(f.col(new_col), '\W+', " "))
rv7 = rv6.withColumn(new_col,f.lower(f.col(new_col)))
return rv7
In [88]:
rv_raw.columns
Out[88]:
In [89]:
usr_clusters2.createOrReplaceTempView('usr_clusters2')
rv_raw.createOrReplaceTempView('rv_raw')
In [90]:
spark.sql('''
SELECT
count(*)
FROM rv_raw
''').show()
In [91]:
words = spark.sql('''
SELECT
A.review_id,
A.business_id,
A.user_id,
A.text,
max(B.text) as text_tp
FROM rv_raw as A
LEFT JOIN tp as B
ON A.user_id = B.user_id and A.business_id = B.business_id
GROUP BY 1, 2, 3, 4
''')
In [92]:
words.createOrReplaceTempView('words')
In [93]:
spark.sql('''
SELECT
count(*)
FROM words
''').show()
In [94]:
bz.columns
Out[94]:
In [95]:
words2 = spark.sql('''
SELECT A.*
FROM bz22 as B
LEFT JOIN words as A
ON A.business_id = B.business_id
''')
In [96]:
words2.count()
Out[96]:
In [97]:
words2.createOrReplaceTempView('words2')
In [99]:
words3 = spark.sql('''
SELECT A.*,
B.cluster_usr
FROM words2 as A
LEFT JOIN usr_clusters2 as B
ON A.user_id = B.user_id
''')
In [100]:
words3.count()
Out[100]:
In [101]:
words4 = word_clean(words3,'text','text_clean')
words5 = word_clean(words4,'text_tp','text_clean_tp')
In [102]:
words5.show(4)
In [105]:
from pyspark.sql.functions import split, regexp_replace, col, split, stddev, regexp_replace
words6 = words5.select("cluster_usr", split(
"text_clean", r"[\s.]+").alias("text2"), split("text_clean_tp", r"[\s.]+").alias("text_tp2"))
In [110]:
locale = sc._jvm.java.util.Locale
locale.setDefault(locale.forLanguageTag("en-US"))
In [111]:
remover = StopWordsRemover(inputCol="text2", outputCol="text_filtered")
words7 = remover.transform(words6)
words7.show(5)
In [112]:
from pyspark.sql.functions import concat_ws
words8 = words7.select("cluster_usr", "text_filtered",
concat_ws(' ', "text_filtered").alias("text_filtered2"))
In [113]:
words9 = words8.withColumn('wordCount', f.size(f.split(f.col('text_filtered2'),' ')))
In [114]:
words10 = words9.withColumn('word', f.explode(f.split(f.col('text_filtered2'), ' ')))\
.groupBy('word','cluster_usr')\
.count()\
.sort('count', ascending = False)
words10.show()
In [115]:
words10.createOrReplaceTempView('words10')
In [116]:
cls = []
for i in tqdm(range(1, 11)):
cls.append(spark.sql(f'''
SELECT word,
count
FROM words10
WHERE word IS NOT NULL
AND cluster_usr == {i}
ORDER BY count DESC limit 50
'''))
cls[i-1] = [tuple(x) for x in cls[i-1].toPandas().to_numpy()]
In [117]:
cluster = 1
for cl in cls:
print(f"Cluster {cluster}")
plt.figure(figsize=(10, 10))
plt.imshow(WordCloud().generate(str(cl).replace("'", "")), interpolation='bilinear')
plt.axis('off')
plt.show()
cluster += 1
Heatmap¶
- Criando mapa de calor da concentração de reviews
In [118]:
base_mapas = spark.sql('''
SELECT A.*, B.latitude, B.longitude
FROM bz22 as B
LEFT JOIN words as A
ON A.business_id = B.business_id
''')
In [119]:
base_mapas.columns
Out[119]:
In [120]:
base_mapas.createOrReplaceTempView('base_mapas')
In [121]:
mapa1 = spark.sql("""
SELECT latitude,
longitude
FROM base_mapas
WHERE latitude is not null
AND longitude is not null
""")
Decobrindo o ponto central de Latitude e Longetude do Mapa
In [122]:
spark.sql("""
SELECT avg(latitude) as avg_lat,
avg(longitude) as avg_long
FROM base_mapas
""").show()
In [123]:
import folium
from folium import plugins
mapa = folium.Map(location=[43.673, -79.391],
zoom_start=11,
tiles='Stamen Toner')
# OpenStreetMap, Stamen Terrain, Stamen Toner
In [124]:
mapa
lat = mapa1.toPandas()['latitude'].values
lon = mapa1.toPandas()['longitude'].values
In [125]:
coordenadas = []
for la, lo in zip(lat, lon):
coordenadas.append([la,lo])
mapa.add_child(plugins.HeatMap(coordenadas))
Out[125]: